package com.weng.starter.kafka.sample1.api;

import com.weng.starter.kafka.sample1.config.KafkaConfig;
import com.weng.starter.kafka.sample1.model.Book;
import com.weng.starter.kafka.sample1.service.BookProducerService;
import io.swagger.annotations.Api;
import io.swagger.annotations.ApiOperation;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaOperations;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;

import java.util.concurrent.atomic.AtomicLong;

@RestController
@RequestMapping(value = "/kafka-sample1")
@Api(value = "kafka测试sample1", tags = {"kafka测试sample1接口"})
public class BookApi {

    @Autowired
    private  BookProducerService producer;

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    private AtomicLong atomicLong = new AtomicLong();

    @GetMapping
    @ApiOperation(value = "发送消息", notes = "")
//    @Transactional(rollbackFor = Exception.class)
    public void sendMessageToKafkaTopic(@RequestParam("name") String name) {
        this.producer.sendMessage(KafkaConfig.MY_TOPTIC_TEST_1, new Book(atomicLong.addAndGet(1), name));
        this.producer.sendMessage(KafkaConfig.MY_TOPTIC_TEST_2, new Book(atomicLong.addAndGet(1), name));
    }

    @ApiOperation(value = "事务测试", notes = "")
    @PostMapping(path = "/transaction/{name}")
    @Transactional(rollbackFor = Exception.class)
    public void sendFoo(@PathVariable String name) {

        this.producer.sendMessage(KafkaConfig.MY_TOPTIC_TEST_1, new Book(atomicLong.addAndGet(1), name));
        // 模拟发生异常，发送异常以后，前面的发送操作就会不成功
        int a = 1 / 0;

    }

    /**
     *  这种方式开启事务是不需要配置事务管理器的，也可以称为本地事务
     * @throws InterruptedException
     */
    @ApiOperation(value = "本地事务测试", notes = "")
    @PostMapping(path = "/transaction-local/{name}")
    public void testExecuteInTransaction(@PathVariable String name) throws InterruptedException {
        kafkaTemplate.executeInTransaction(new KafkaOperations.OperationsCallback() {
            @Override
            public Object doInOperations(KafkaOperations kafkaOperations) {
                kafkaOperations.send(KafkaConfig.MY_TOPTIC_TEST_1, name);
                throw new RuntimeException("fail");
//                return true;
            }
        });
    }
}
